package com.zhou.common.mybatis.template.shiro.zhou.config.mq;


import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import com.zhou.common.mybatis.template.shiro.zhou.exception.BizException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.io.UnsupportedEncodingException;

@Slf4j
@Configuration
@ConditionalOnProperty(prefix = "rocketmq", name = "enable",havingValue = "true")
@EnableConfigurationProperties(RocketMqProperties.class)
public class LocalRocketMqConfig {

    @Autowired
    private  RocketMqProperties rocketMqProperties;

    @Bean(initMethod = "start", destroyMethod = "shutdown")
    @ConditionalOnProperty(prefix = "rocketmq.producer", name = "enable",havingValue = "true")
    public DefaultMQProducer defaultProducer() throws MQClientException {
        DefaultMQProducer producer = new DefaultMQProducer(rocketMqProperties.getProducer().getGroup());
        producer.setNamesrvAddr(rocketMqProperties.getNameServer());
        producer.setVipChannelEnabled(false);
        producer.setMaxMessageSize(rocketMqProperties.getProducer().getMaxMessageSize());
        producer.setSendMsgTimeout(rocketMqProperties.getProducer().getSendMessagetimeout());
        producer.setRetryTimesWhenSendAsyncFailed(rocketMqProperties.getProducer().getRetryTimesWhenSendAsyncFailed());
        producer.setCompressMsgBodyOverHowmuch(rocketMqProperties.getProducer().getCompressMessageBodyThreshold());
        producer.setRetryTimesWhenSendFailed(rocketMqProperties.getProducer().getRetryTimesWhenSendFailed());
        return producer;
    }

    @Bean(initMethod = "start", destroyMethod = "shutdown")
    @ConditionalOnProperty(prefix = "rocketmq.consumer", name = "enable",havingValue = "true")
    public DefaultMQPushConsumer defaultConsumer() throws MQClientException {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(rocketMqProperties.getConsumer().getGroup());
        consumer.setNamesrvAddr(rocketMqProperties.getNameServer());
        /*consumer.setConsumeThreadMin(consumeThreadMin);
        consumer.setConsumeThreadMax(consumeThreadMax);
        consumer.setConsumeMessageBatchMaxSize(consumeMessageBatchMaxSize);*/
        consumer.subscribe(rocketMqProperties.getConsumer().getCommonTopic(), "*");
        /**
         * 设置consumer第一次启动是从队列头部开始还是队列尾部开始
         * 如果不是第一次启动，那么按照上次消费的位置继续消费
         */
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);

        consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
            // msgs中只收集同一个topic，同一个tag，并且key相同的message
            // 会把不同的消息分别放置到不同的队列中
            try {
                for (MessageExt msg : msgs) {
                    log.info("body={}",new String(msg.getBody(), "utf-8"));
                    //消费者获取消息 这里只输出 不做后面逻辑处理
//                    String body = new String(msg.getBody(), "utf-8");
                    String body = new String(msg.getBody());
                    log.info("Consumer-获取消息-主题topic为={}, 消费消息为={}", msg.getTopic(), body);
                }
                if (1==1){
                    throw new BizException(99,"测试");
                }
            } catch (UnsupportedEncodingException e) {
                log.error("",e);
            }
            return ConsumeConcurrentlyStatus.RECONSUME_LATER;
        });

        return consumer;
    }

}
